package com.aliyun.adb.contest;




import java.io.*;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;



/***
 * 
 * title: 高效亿级txt文件 切分
 [1000000000000000000, 1913708004094975089, 2827416008189950178, 3741124012284925267, 4654832016379900356,
 5568540020474875445, 6482248024569850534, 7395956028664825623, 8309664032759800712, 9223372036854775807]
 * @author Geng
 * @date 2021-5-19 11:47:37
 */
public class SplitBigFileTools {



    public static long[] CNT_LINE_ONE;
    public static long[] CNT_LINE_TWO;

    public static long[] SUM_ONE;
    public static long[] SUM_TWO;

    public static String LINE_SEP = "\n";

    public static BlockingQueue<FileNode> blockingQueue1;
    public static BlockingQueue<FileNode> blockingQueue2;

    public AtomicInteger total_line1 = new AtomicInteger(0);
    public AtomicInteger total_line2 = new AtomicInteger(0);

    public static String WORK_PATH;

    //cnt line
    public static AtomicInteger[] LINE_ONE;
    public static AtomicInteger[] LINE_TWO;


    public static Set<StartEndPair> startEndPairs;

    private static RandomAccessFile rAccessFile;


    static CountDownLatch countDownLatch;

    static BufferedWriter[] bws1;
    static BufferedWriter[] bws2;

    static ExecutorService writeService1;
    static ExecutorService writeService2;

    private StringBuilder[] sb1;
    private StringBuilder[] sb2;

    private byte[][] mem1;
    private byte[][] mem2;

    int[] arrIdx1;
    int[] arrIdx2;

    private long[] offset1;
    private long[] offset2;

    public SplitBigFileTools() {
        String os = System.getProperties().getProperty("os.name");
        if(os.indexOf("Windows") >= 0){
            Const.LINE_NUM = 10000;
            countDownLatch = new CountDownLatch(20000);
            Const.LINE_FLUSH_CNT = 5000;
        }
    }

    public static void main(String[] args) {
        String s = "513690303820996718";
        int num = 0;
        byte[] bytes = s.getBytes();
        System.out.println(bytes.length);
    }



    /**
     * 不进行切分，单线程读，传统方式单线程读，mmp多线程写
     * @param workPath
     * @throws IOException
     */
    public void read(File fileName,String workPath){
        //readAndWriteFileIoBuffer(fileName,workPath);
       // readAndWriteFileIoBuffer(fileName,workPath);
        //testWhich(fileName);
        readAndWriteFileIoFCV2(fileName,workPath);
    }

    /*
        split and producer and comsumer
        abandon
     */
    public void readV2(File fileName,String workPath){
        WORK_PATH = workPath;
        produceAndConsumeLine(fileName);
    }

    /*
        mmp单线程读，mmp多线程写
     */
    public void readV3(File fileName,String workPath){
        readAndWriteMMPV2(fileName, workPath);
    }

    public void readAndWriteMMP(File fileName,String workPath){
        WORK_PATH = workPath;
        int len = Const.WRITE_FILE_NUM;

        startEndPairs = new HashSet<>();
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];


        for(int i = 0;i < len;i++){
            fileNames1[i] = WORK_PATH + "/" + i + "output1";
            fileNames2[i] = WORK_PATH + "/" + i + "output2";
        }

        sb1 = new StringBuilder[len];
        sb2 = new StringBuilder[len];
        offset1 = new long[len];
        offset2 = new long[len];

        //init buffer
        for(int i = 0;i < len;i++){
            sb1[i] = new StringBuilder(Const.STR_CAPACITY);
            sb2[i] = new StringBuilder(Const.STR_CAPACITY);
        }

        ExecutorService service = Executors.newFixedThreadPool(Const.THREAD_POOL_NUM);


        FileChannel[] fcs1 = new FileChannel[len];
        FileChannel[] fcs2 = new FileChannel[len];

        for(int i = 0;i < len;i++){
            try {
                fcs1[i] = new RandomAccessFile(fileNames1[i],"rw").getChannel();
                fcs2[i] = new RandomAccessFile(fileNames2[i],"rw").getChannel();
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }


        FileInputStream fis = null;
        FileChannel fc = null;

        int cntLine = 0;
        long l1 = System.currentTimeMillis();
        try {
            fis = new FileInputStream(fileName);
            fc = fis.getChannel();

            long fileSize = fileName.length();
            //  System.out.println(fileSize);
            long everySize = fileSize/Const.SPLIT_FILE_NUM;
            rAccessFile = new RandomAccessFile(fileName,"r");
            calculateStartEnd(0,everySize,fileSize);
            //byte[] tmp_byte = new byte[100];
            long t  = 0;
            //分批次读
            for(StartEndPair pair : startEndPairs){
                MappedByteBuffer bf = fc.map(FileChannel.MapMode.READ_ONLY, pair.start, pair.end-pair.start+1);
                long time = System.currentTimeMillis();
                while(bf.hasRemaining()){
                    byte b = bf.get();
                    if(b == '\n' || b == '\r'){
                        if(t == 0)  continue;
                        int whichFile = getWhichFile(t);
                        if(Const.USE_OFFSET){
                            if(whichFile != 0) t -= Const.COM_MAP[whichFile-1];
                        }
                        sb2[whichFile].append(t+LINE_SEP);
                        CNT_LINE_TWO[whichFile]++;
                        t = 0;
                        cntLine++;
                    }else if(b == Const.COMMA){
                        if(t == 0)  continue;
                        int whichFile = getWhichFile(t);
                        if(Const.USE_OFFSET){
                            if(whichFile != 0) t -= Const.COM_MAP[whichFile-1];
                        }
                        sb1[whichFile].append(t+LINE_SEP);
                        t = 0;
                        CNT_LINE_ONE[whichFile]++;
                    } else{
                        if(b >= 48 && b <= 57){
                            t = t*10 + (b-48);
                        }
                    }
                }
                System.out.println("read and write once spend: " + (System.currentTimeMillis()-time));
                flush(fcs1,fcs2 ,service);
            }
            System.out.println("total line: "+cntLine);
            System.out.println("read and write spend: " + (System.currentTimeMillis()-l1));

            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }

            printLog();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            closeFC(fcs1,fcs2);
            service.shutdown();
        }
    }

    public void readAndWriteMMPV2(File fileName,String workPath){
        WORK_PATH = workPath;
        int len = Const.WRITE_FILE_NUM;

        startEndPairs = new HashSet<>();
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        mem1 = new byte[len][Const.BYTE_LEN];
        mem2 = new byte[len][Const.BYTE_LEN];

        for(int i = 0;i < len;i++){
            fileNames1[i] = WORK_PATH + "/" + i + "output1";
            fileNames2[i] = WORK_PATH + "/" + i + "output2";
        }

        offset1 = new long[len];
        offset2 = new long[len];

        arrIdx1 = new int[len];
        arrIdx2 = new int[len];



        ExecutorService service = Executors.newFixedThreadPool(Const.THREAD_POOL_NUM);


        FileChannel[] fcs1 = new FileChannel[len];
        FileChannel[] fcs2 = new FileChannel[len];

        for(int i = 0;i < len;i++){
            try {
                fcs1[i] = new RandomAccessFile(fileNames1[i],"rw").getChannel();
                fcs2[i] = new RandomAccessFile(fileNames2[i],"rw").getChannel();
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }


        FileInputStream fis = null;
        FileChannel fc = null;

        int cntLine = 0;
        long l1 = System.currentTimeMillis();
        try {
            fis = new FileInputStream(fileName);
            fc = fis.getChannel();

            long fileSize = fileName.length();
            //  System.out.println(fileSize);
            long everySize = fileSize/Const.SPLIT_FILE_NUM;
            rAccessFile = new RandomAccessFile(fileName,"r");
            calculateStartEnd(0,everySize,fileSize);

            //
            //byte[] tmp_byte = new byte[100];
            long t  = 0;
            byte[] tmp_bytes = new byte[100];
            int tmp_index = 0;
            int whichFile = 0;
            int whichFile18 = 0;
            //分批次读
            for(StartEndPair pair : startEndPairs){
                MappedByteBuffer bf = fc.map(FileChannel.MapMode.READ_ONLY, pair.start, pair.end-pair.start+1);
                long time = System.currentTimeMillis();
                while(bf.hasRemaining()){
                    byte b = bf.get();
                    if(b == '\n' || b == '\r'){
                        if(t == 0)  continue;
                      //  int whichFile = getWhichFile(t);
                        if(tmp_index < 18) whichFile = 0;
                        else if(tmp_index == 18) whichFile = whichFile18;
                        else whichFile += 1;
                        tmp_bytes[tmp_index++] = 10;
                        System.arraycopy(tmp_bytes,0 , mem2[whichFile], arrIdx2[whichFile], tmp_index);
                        arrIdx2[whichFile] += tmp_index;
                        CNT_LINE_TWO[whichFile]++;
                        t = 0;tmp_index = 0;
                        cntLine++;
                    }else if(b == Const.COMMA){
                        if(t == 0)  continue;
                       // int whichFile = getWhichFile(t);
                        if(tmp_index < 18) whichFile = 0;
                        else if(tmp_index == 18) whichFile = whichFile18;
                        else whichFile += 1;
                        tmp_bytes[tmp_index++] = 10;
                        System.arraycopy(tmp_bytes,0 , mem1[whichFile], arrIdx1[whichFile], tmp_index);
                        arrIdx1[whichFile] += tmp_index;
                        t = 0;tmp_index = 0;
                        CNT_LINE_ONE[whichFile]++;
                    } else{
                        if(b >= 48 && b <= 57){
                            t = t*10 + (b-48);
                            if(t < 100){
                                whichFile = (int)t;
                            }
                            if(t < 10){
                                whichFile18 = (int)t;
                            }
                            tmp_bytes[tmp_index++] = b;
                        }
                    }
                }
                System.out.println("read and write once spend: " + (System.currentTimeMillis()-time));
                flushV2(fcs1,fcs2 ,service);
            }
            System.out.println("total line: "+cntLine);
            System.out.println("read and write spend: " + (System.currentTimeMillis()-l1));

            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }
            printLog();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            closeFC(fcs1,fcs2);
            service.shutdown();
        }
    }

    /*
      进行数据切分，按照大小，分为10*2个文件 use BufferedWriter
    */
    public void readAndWriteFileIoBuffer(File file,String workPath) {
        BufferedReader reader = null;
        int len = Const.COM_MAP.length;
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        for(int i = 0;i < len;i++){
            fileNames1[i] = workPath + "/" + i + "output1";
            fileNames2[i] = workPath + "/" + i + "output2";
        }
        //here can join buffer
        bws1 = new BufferedWriter[len];
        bws2 = new BufferedWriter[len];
        for(int i = 0;i < len;i++){
            try {
                bws1[i] = new BufferedWriter(new FileWriter(fileNames1[i]),Const.WRITER_BUFFER_SIZE);
                bws2[i] = new BufferedWriter(new FileWriter(fileNames2[i]),Const.WRITER_BUFFER_SIZE);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            long l1 = System.currentTimeMillis();
            reader = new BufferedReader(new FileReader(file),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            String line = null;
            long cntLine = 0;
            while((line = reader.readLine()) != null){
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");
                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);
                int file1 = getWhichFile(num1),file2 = getWhichFile(num2);
               // int file1 = new Random().nextInt(len),file2 = new Random().nextInt(len);
                bws1[file1].write(num1+"");bws2[file2].write(num2+"");
                bws1[file1].newLine();bws2[file2].newLine();
                CNT_LINE_ONE[file1]++;
                CNT_LINE_TWO[file2]++;
                cntLine++;
            }
            System.out.println("total line: "+cntLine);
            System.out.println("read delay: " + (System.currentTimeMillis()-l1));
            System.out.println(Arrays.toString(CNT_LINE_ONE));
            System.out.println(Arrays.toString(CNT_LINE_TWO));
            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            closeBW(bws1,bws2);
        }
    }

    /*
    one thread read and some thead write
     */
    public void readAndWriteFileIoBufferV2(File file,String workPath) {
        BufferedReader reader = null;
        int len = Const.COM_MAP.length;
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        for(int i = 0;i < len;i++){
            fileNames1[i] = workPath + "/" + i + "output1";
            fileNames2[i] = workPath + "/" + i + "output2";
        }
        //here can join buffer
        bws1 = new BufferedWriter[len];
        bws2 = new BufferedWriter[len];



        for(int i = 0;i < len;i++){
            try {
                bws1[i] = new BufferedWriter(new FileWriter(fileNames1[i]),Const.WRITER_BUFFER_SIZE);
                bws2[i] = new BufferedWriter(new FileWriter(fileNames2[i]),Const.WRITER_BUFFER_SIZE);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        //start write file
        startWriteThread();

        try {
            long l1 = System.currentTimeMillis();
            reader = new BufferedReader(new FileReader(file),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            String line = null;
            long cntLine = 0;
            while((line = reader.readLine()) != null){
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");

                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);
                int file1 = getWhichFile(num1),file2 = getWhichFile(num2);
                blockingQueue1.put(new FileNode(num1,file1));
                blockingQueue2.put(new FileNode(num2,file2));
                CNT_LINE_ONE[file1]++;
                CNT_LINE_TWO[file2]++;
                cntLine++;
            }

            //wait to write
            countDownLatch.await();

            closeBW();
            closeTheadService();

            System.out.println("total line: "+cntLine);
            System.out.println("read and write file delay: " + (System.currentTimeMillis()-l1));
            System.out.println(Arrays.toString(CNT_LINE_ONE));
            System.out.println(Arrays.toString(CNT_LINE_TWO));
            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /*
       use  BufferedOutputStream
     */
    public void readAndWriteFileIoBos(File file,String workPath) {
        BufferedReader reader = null;
        int len = Const.COM_MAP.length;
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        for(int i = 0;i < len;i++){
            fileNames1[i] = workPath + "/" + i + "output1";
            fileNames2[i] = workPath + "/" + i + "output2";
        }

        //here can join buffer
        BufferedOutputStream[] bos1 = new BufferedOutputStream[len];
        BufferedOutputStream[] bos2 = new BufferedOutputStream[len];

        for(int i = 0;i < len;i++){
            try {
                bos1[i] = new BufferedOutputStream(new FileOutputStream(fileNames1[i]));
                bos2[i] = new BufferedOutputStream(new FileOutputStream(fileNames2[i]));
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }
        long l1 = System.currentTimeMillis();
        long cntLine = 0;
        String line;
        try{
            reader = new BufferedReader(new FileReader(file),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            while((line = reader.readLine()) != null){
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");
                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);
                int file1 = getWhichFile(num1),file2 = getWhichFile(num2);
                bos1[file1].write((num1+LINE_SEP).getBytes());bos2[file2].write((num2+LINE_SEP).getBytes());
                CNT_LINE_ONE[file1]++;
                CNT_LINE_TWO[file2]++;
                cntLine++;
            }
            System.out.println("total line: "+cntLine);
            System.out.println("read delay: " + (System.currentTimeMillis()-l1));
            System.out.println(Arrays.toString(CNT_LINE_ONE));
            System.out.println(Arrays.toString(CNT_LINE_TWO));
            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            closeBos(bos1,bos2);
        }
    }


    /*
      use nio bytebuffer
     */
    public void readAndWriteFileIoFC(File file,String workPath) {
        BufferedReader reader = null;
        int len = Const.COM_MAP.length;
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        for(int i = 0;i < len;i++){
            fileNames1[i] = workPath + "/" + i + "output1";
            fileNames2[i] = workPath + "/" + i + "output2";
        }

        //here can join buffer
        ByteBuffer[] bf1 = new ByteBuffer[len];
        ByteBuffer[] bf2 = new ByteBuffer[len];

        //init buffer
        for(int i = 0;i < len;i++){
            bf1[i] = ByteBuffer.allocate(Const.WRITER_BUFFER_SIZE);
            bf2[i] = ByteBuffer.allocate(Const.WRITER_BUFFER_SIZE);
        }

        FileChannel[] fcs1 = new FileChannel[len];
        FileChannel[] fcs2 = new FileChannel[len];

        for(int i = 0;i < len;i++){
            try {
                fcs1[i] = new RandomAccessFile(fileNames1[i],"rw").getChannel();
                fcs2[i] = new RandomAccessFile(fileNames2[i],"rw").getChannel();
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }
        long l1 = System.currentTimeMillis();
        long cntLine = 0;
        String line;
        try{
            reader = new BufferedReader(new FileReader(file),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            while((line = reader.readLine()) != null){
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");
                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);
              //  System.out.println(num1 + " "+num2);
                int file1 = getWhichFile(num1),file2 = getWhichFile(num2);
                byte[] bytes1 = String.valueOf(num1+LINE_SEP).getBytes(),bytes2 = String.valueOf(num2+LINE_SEP).getBytes();
                //buffer is no use
                if(bytes1.length > bf1[file1].remaining()){
                    fcs1[file1].write(bf1[file1]);
                    bf1[file1].flip();
                    while(bf1[file1].hasRemaining()){
                        fcs1[file1].write(bf1[file1]);
                    }
                    bf1[file1].clear();
                }
                bf1[file1].put(bytes1);

                if(bytes2.length > bf2[file2].remaining()){
                    bf2[file2].flip();
                    while(bf2[file2].hasRemaining()){
                        fcs2[file2].write(bf2[file2]);
                    }
                    bf2[file2].clear();
                }
                bf2[file2].put(bytes2);

                CNT_LINE_ONE[file1]++;
                CNT_LINE_TWO[file2]++;
                cntLine++;
            }
            for(int i = 0;i < len;i++){
                bf1[i].flip();
                while(bf1[i].hasRemaining()){
                    fcs1[i].write(bf1[i]);
                }
                bf2[i].flip();
                while(bf2[i].hasRemaining()){
                    fcs2[i].write(bf2[i]);
                }
            }
            System.out.println("total line: "+cntLine);
            System.out.println("read delay: " + (System.currentTimeMillis()-l1));
            System.out.println(Arrays.toString(CNT_LINE_ONE));
            System.out.println(Arrays.toString(CNT_LINE_TWO));
            int length = CNT_LINE_ONE.length;
            SUM_ONE = new long[length];
            SUM_TWO = new long[length];
            SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
            for(int i = 1;i < length;i++){
                SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
                SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            closeFC(fcs1,fcs2);
        }
    }

    public void readAndWriteFileIoFCV2(File file,String workPath) {
        BufferedReader reader = null;
        int len = Const.WRITE_FILE_NUM;
        String[] fileNames1 = new String[len];
        String[] fileNames2 = new String[len];

        //这个要用全局
        CNT_LINE_ONE = new long[len];
        CNT_LINE_TWO = new long[len];

        for(int i = 0;i < len;i++){
            fileNames1[i] = workPath + "/" + i + "output1";
            fileNames2[i] = workPath + "/" + i + "output2";
        }

        sb1 = new StringBuilder[len];
        sb2 = new StringBuilder[len];
        offset1 = new long[len];
        offset2 = new long[len];

        //init buffer
        for(int i = 0;i < len;i++){
            sb1[i] = new StringBuilder();
            sb2[i] = new StringBuilder();
        }

        ExecutorService service = Executors.newFixedThreadPool(Const.THREAD_POOL_NUM);


        FileChannel[] fcs1 = new FileChannel[len];
        FileChannel[] fcs2 = new FileChannel[len];
        try {
            for(int i = 0;i < len;i++){
                fcs1[i] = new RandomAccessFile(fileNames1[i],"rw").getChannel();
                fcs2[i] = new RandomAccessFile(fileNames2[i],"rw").getChannel();
            }
         } catch (FileNotFoundException e) {
            e.printStackTrace();
        }

       // System.out.println("init fcs spend:" + (System.currentTimeMillis()-tInit));

        long l1 = System.currentTimeMillis();
        long cntLine = 0;
        String line;
        try{
            reader = new BufferedReader(new FileReader(file),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            while((line = reader.readLine()) != null){
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");
                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);
                int file1 = getWhichFile(num1),file2 = getWhichFile(num2);
                if(Const.USE_OFFSET){
                    if(file1 != 0) num1 -= Const.COM_MAP[file1-1];
                    if(file2 != 0) num2 -= Const.COM_MAP[file2-1];
                }
                sb1[file1].append(num1+LINE_SEP);sb2[file2].append(num2+LINE_SEP);
                CNT_LINE_ONE[file1]++; CNT_LINE_TWO[file2]++;
                cntLine++;
                if(cntLine % Const.LINE_FLUSH_CNT == 0){
                    //flush(fcs1,fcs2);
                     flush(fcs1,fcs2,service);
                }
            }

            System.out.println("total line: "+cntLine);
            System.out.println("read and write spend: " + (System.currentTimeMillis()-l1));

            printLog();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            closeFC(fcs1,fcs2);
            service.shutdown();
        }
    }

    public void printLog(){
        System.out.println(Arrays.toString(CNT_LINE_ONE));
        System.out.println(Arrays.toString(CNT_LINE_TWO));
        int length = CNT_LINE_ONE.length;
        SUM_ONE = new long[length];
        SUM_TWO = new long[length];
        SUM_ONE[0] = CNT_LINE_ONE[0];SUM_TWO[0] = CNT_LINE_TWO[0];
        for(int i = 1;i < length;i++){
            SUM_ONE[i] = SUM_ONE[i-1] + CNT_LINE_ONE[i];
            SUM_TWO[i] = SUM_TWO[i-1] + CNT_LINE_TWO[i];
        }

        System.out.println(Arrays.toString(SUM_ONE));
        System.out.println(Arrays.toString(SUM_TWO));
    }

    public void flush(FileChannel[] fc1,FileChannel[] fc2)  {
        try {
            for(int i = 0;i < fc1.length;i++){
                byte[] bs1 = sb1[i].toString().getBytes();byte[] bs2 = sb2[i].toString().getBytes();
                MappedByteBuffer mf1 = fc1[i].map(FileChannel.MapMode.READ_WRITE, offset1[i], bs1.length);
                MappedByteBuffer mf2 = fc2[i].map(FileChannel.MapMode.READ_WRITE, offset2[i], bs2.length);
                mf1.put(bs1); mf2.put(bs2);
                offset1[i] += bs1.length; offset2[i] += bs2.length;
                sb1[i].delete(0,sb1[i].length()); sb2[i].delete(0,sb2[i].length());
            }

        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    //some thread to flush2?
    public void flush(FileChannel[] fc1,FileChannel[] fc2,ExecutorService service)  {
        long t1 = System.currentTimeMillis();
        AtomicInteger incr = new AtomicInteger(0);
        CountDownLatch downLatch = new CountDownLatch(Const.WRITE_FILE_NUM);
        for (int j = 0; j < Const.WRITE_FILE_NUM; j++) {
            service.execute(() -> {
                try {
                    int i = incr.getAndIncrement();
                    byte[] bs1 = sb1[i].toString().getBytes();byte[] bs2 = sb2[i].toString().getBytes();
                    MappedByteBuffer mf1 = fc1[i].map(FileChannel.MapMode.READ_WRITE, offset1[i], bs1.length);
                    MappedByteBuffer mf2 = fc2[i].map(FileChannel.MapMode.READ_WRITE, offset2[i], bs2.length);
                    // System.out.println(mf1 + " " +mf2);
                    mf1.put(bs1);mf2.put(bs2);
                    offset1[i] += bs1.length;offset2[i] += bs2.length;
                    sb1[i].delete(0,sb1[i].length());sb2[i].delete(0,sb2[i].length());
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    downLatch.countDown();
                }
            });
        }
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("flush spend: "+(System.currentTimeMillis()-t1));
    }

    public void flushV2(FileChannel[] fc1,FileChannel[] fc2,ExecutorService service)  {
        long t1 = System.currentTimeMillis();
        AtomicInteger incr = new AtomicInteger(0);
        //delete 9 files
        CountDownLatch downLatch = new CountDownLatch(Const.WRITE_FILE_NUM);
        for (int j = 0; j < Const.WRITE_FILE_NUM; j++) {
            service.execute(() -> {
                try {
                    int i = incr.getAndIncrement();
                    MappedByteBuffer mf1 = fc1[i].map(FileChannel.MapMode.READ_WRITE, offset1[i], arrIdx1[i]);
                    MappedByteBuffer mf2 = fc2[i].map(FileChannel.MapMode.READ_WRITE, offset2[i], arrIdx2[i]);
                    // System.out.println(mf1 + " " +mf2);
                    mf1.put(mem1[i], 0, arrIdx1[i]); mf2.put(mem2[i],0,arrIdx2[i]);
                    offset1[i] += arrIdx1[i];offset2[i] += arrIdx2[i];
                    arrIdx1[i] = 0;arrIdx2[i] = 0;
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    downLatch.countDown();
                }
            });
        }
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("flush spend: "+(System.currentTimeMillis()-t1));
    }

    //


    public void closeBos(BufferedOutputStream[] bos1,BufferedOutputStream[] bos2){
        for(int i = 0;i < bos1.length;i++){
            try {
                bos1[i].flush();bos2[i].flush();
                bos1[i].close();bos2[i].close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void closeBW(BufferedWriter[] bws1,BufferedWriter[] bws2){
        for(int i = 0;i < bws1.length;i++){
            try {
                bws1[i].flush();
                bws1[i].close();
                bws2[i].flush();
                bws2[i].close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public void closeFC(FileChannel[] fcs1, FileChannel[] fcs2){
        for(int i = 0;i < fcs1.length;i++){
            try {
                fcs1[i].close();
                fcs2[i].close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    //二分优化一下吧
    public int getWhichFile(long num){
        int l = 0,r = Const.COM_MAP.length-1;
        while(l < r){
            int mid = (l+r)/2;
            if(num <= Const.COM_MAP[mid]) r = mid;
            else l = mid + 1;
        }
        return l;
    }

    /*
    mmp write file
     */
    public static void writeTreeSetToFileV2(String filePath, List<Long> list1, List<Long> list2) {
        if(list1.size() < 1) return;
        Collections.sort(list1);
        Collections.sort(list2);
        RandomAccessFile acf;
        FileChannel fc = null;
        try {
            String fName = filePath;
            acf = new RandomAccessFile(fName, "rw");
            fc = acf.getChannel();
            int offset = 0;
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < list1.size(); i++) {
                sb.append(list1.get(i) + "," + list2.get(i) + "\n");
            }
            byte[] bs = sb.toString().getBytes();
            MappedByteBuffer mbuf = fc.map(FileChannel.MapMode.READ_WRITE, offset, bs.length);
            mbuf.put(bs);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                fc.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /*
      producer and consumer module read file
     */


    public void splitFileV2(File filePath) throws IOException {
        //System.out.println(filePath.getName());
        FileInputStream fis = new FileInputStream(filePath);
        FileChannel inputChannel = fis.getChannel();
        long fileSize = filePath.length();
        //  System.out.println(fileSize);
        long everySize = fileSize/Const.SPLIT_FILE_NUM;
        rAccessFile = new RandomAccessFile(filePath,"r");
        calculateStartEnd(0,everySize,fileSize);
        int i = 0;
        for(StartEndPair pair:startEndPairs){
            System.out.println("split file pair: "+pair);
            String name = WORK_PATH + "/" +filePath.getName()+(++i);
            FileOutputStream fos = new FileOutputStream(name);
            FileChannel outputChannel = fos.getChannel();
            inputChannel.transferTo(pair.start, pair.end-pair.start+1, outputChannel);// 通道传输文件数据
            outputChannel.close();
            fos.close();
        }
        inputChannel.close();
        fis.close();
    }

    public  void calculateStartEnd(long start,long size,long fileLength) throws IOException{
        if(start>fileLength-1){
            return;
        }
        StartEndPair pair = new StartEndPair();
        pair.start=start;
        long endPosition = start+size-1;
        if(endPosition>=fileLength-1){
            pair.end=fileLength-1;
            startEndPairs.add(pair);
            return;
        }
        rAccessFile.seek(endPosition);
        byte tmp =(byte) rAccessFile.read();

        while(tmp!='\n' && tmp!='\r'){
            endPosition++;
            if(endPosition>=fileLength-1){
                endPosition=fileLength-1;
                break;
            }
            rAccessFile.seek(endPosition);
            tmp =(byte) rAccessFile.read();
        }
        pair.end=endPosition;
        startEndPairs.add(pair);

        calculateStartEnd(endPosition+1, size,fileLength);
    }

    public void produceAndConsumeLine(File filePath){
        int segment = Const.SPLIT_FILE_NUM;
        int writeFileNum = Const.WRITE_FILE_NUM;
        try {
            long t1 = System.currentTimeMillis();
            splitFileV2(filePath);
            System.out.println("splitFile spend: "+ (System.currentTimeMillis()-t1));
        } catch (IOException e) {
            e.printStackTrace();
        }
        ExecutorService readService = Executors.newFixedThreadPool(segment);
        AtomicInteger incr = new AtomicInteger(1);

        String[] fileNames1 = new String[writeFileNum];
        String[] fileNames2 = new String[writeFileNum];

        for(int i = 0;i < writeFileNum;i++){
            fileNames1[i] = WORK_PATH + "/" + i + "output1";
            fileNames2[i] = WORK_PATH + "/" + i + "output2";
        }
        //init BufferedWriter
        bws1 = new BufferedWriter[writeFileNum];
        bws2 = new BufferedWriter[writeFileNum];

        //write threadpool
        for(int i = 0;i < writeFileNum;i++){
            try {
                bws1[i] = new BufferedWriter(new FileWriter(fileNames1[i]));
                bws2[i] = new BufferedWriter(new FileWriter(fileNames2[i]));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        long t = System.currentTimeMillis();
        //read from queue
        //System.out.println("****" + Const.LINE_NUM);
        startWriteThread();

        for (int i = 1; i <= segment; i++) {
            readService.execute(() -> {

                readSmallFile2Queue(WORK_PATH +"/"+ filePath.getName() + incr.getAndIncrement());

            });
        }

       // System.out.println("total_line1:  "+total_line1);
       // System.out.println("total_line2:  "+total_line2);
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        closeBW();
        closeTheadService(readService);
        System.out.println("write file spend: "+(System.currentTimeMillis()-t));

        //calcu Sum
        int length = LINE_ONE.length;
        SUM_ONE = new long[length];
        SUM_TWO = new long[length];

        long[] printOne = new long[length];
        long[] printTwo = new long[length];
        for(int i = 0;i < length;i++){
            printOne[i] = LINE_ONE[i].get();
            printTwo[i] = LINE_TWO[i].get();
        }
        SUM_ONE[0] = LINE_ONE[0].get();SUM_TWO[0] = LINE_TWO[0].get();
        for(int i = 1;i < length;i++){
            SUM_ONE[i] = SUM_ONE[i-1] + LINE_ONE[i].get();
            SUM_TWO[i] = SUM_TWO[i-1] + LINE_TWO[i].get();
        }
        System.out.println("printOne: " + Arrays.toString(printOne));
        System.out.println("printTwo: " + Arrays.toString(printTwo));
        System.out.println("SUM_ONE: " + Arrays.toString(SUM_ONE));
        System.out.println("SUM_TWO: " + Arrays.toString(SUM_TWO));
    }

    public void closeTheadService(ExecutorService readService){
        readService.shutdown();
        writeService1.shutdown();
        writeService2.shutdown();
    }

    public void closeTheadService(){
        writeService1.shutdown();
        writeService2.shutdown();
    }

    public void startWriteThread(){
        AtomicBoolean isFistFile = new AtomicBoolean();
        AtomicBoolean isSecondFile = new AtomicBoolean();
        isFistFile.getAndSet(true);

        //System.out.println(isFistFile.get()+"  " + isSecondFile.get());

        writeService1 = Executors.newFixedThreadPool(Const.WRITE_FILE_NUM);
        writeService2 = Executors.newFixedThreadPool(Const.WRITE_FILE_NUM);

        for (int i = 0; i < Const.LINE_NUM; i++) {
            writeService1.execute(() -> {
                try {
                    write2SmallFromQueue(blockingQueue1,isFistFile.get());
                } finally {
                    countDownLatch.countDown();
                }
            });


            writeService2.execute(() -> {
                try {
                    write2SmallFromQueue(blockingQueue2,isSecondFile.get());
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

    }
    //write to queue
    public void readSmallFile2Queue(String filePath) {
        File file = new File(filePath);
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader(file), (int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
            String line = null;
            while ((line = reader.readLine()) != null) {
                //handle
                if(line.length() == 0 || !Character.isDigit(line.charAt(0))) continue;
                String[] strs = line.split(",");
                //System.out.println(line);
                long num1 = Long.parseLong(strs[0]),num2 = Long.parseLong(strs[1]);

                //write to which file
                int f1 = getWhichFile(num1),f2 = getWhichFile(num2);
                //write to queue;

                blockingQueue1.put(new FileNode(num1,f1));blockingQueue2.put(new FileNode(num2,f2));
                System.out.println("blockingQueue1 size: " + blockingQueue1.size());
                System.out.println("blockingQueue2 size: " + blockingQueue2.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public  void write2SmallFromQueue(BlockingQueue<FileNode> blockingQueue,boolean isFirstFile) {
        String fileName;
        FileNode fileNode = null;
        int fileIndex = 0;
        try {
            fileNode = blockingQueue.take();
            String data = fileNode.data_num+"";
            fileIndex = fileNode.file_num;
            //System.out.println(fileNode);
            if(isFirstFile){
                try {
                    LINE_ONE[fileIndex].getAndIncrement();
                    //System.out.println("1  "+data);
                    synchronized(bws1[fileIndex]){
                        bws1[fileIndex].write(data);bws1[fileIndex].newLine();
                    }
                    total_line1.getAndIncrement();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }else{
                try {
                    LINE_TWO[fileIndex].getAndIncrement();
                    //System.out.println("2***  "+data);
                    synchronized(bws1[fileIndex]){
                        bws2[fileIndex].write(data);bws2[fileIndex].newLine();
                    }
                    total_line2.getAndIncrement();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            //when to close bw todo

        }
    }
    public void closeBW(){
        for(int i = 0;i < bws1.length;i++){
            try {
                bws1[i].flush();
                bws1[i].close();
                bws2[i].flush();
                bws2[i].close();
            } catch (IOException e) {
                    e.printStackTrace();
               }
            }
        System.out.println("close BW complete");
    }

    class StartEndPair{
        public long start;
        public long end;

        @Override
        public String toString() {
            return "start="+start+";end="+end;
        }

        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + (int) (end ^ (end >>> 32));
            result = prime * result + (int) (start ^ (start >>> 32));
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            StartEndPair other = (StartEndPair) obj;
            if (end != other.end)
                return false;
            if (start != other.start)
                return false;
            return true;
        }

    }
    class FileNode{
        public Long data_num;
        public int file_num;

        public FileNode(Long data_num, int file_num) {
            this.data_num = data_num;
            this.file_num = file_num;
        }

        @Override
        public String toString() {
            return "FileNode{" +
                    "data_num=" + data_num +
                    ", file_num=" + file_num +
                    '}';
        }


    }
}

